package com.wsjj.gmall;

import com.wsjj.gmall.base.BaseSQLApp;
import com.wsjj.gmall.constant.Constant;
import com.wsjj.gmall.util.SQLUtil;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 *
 * 加购事务事实表：
 *
 * 加购逻辑：  cart_info
 *          购物车表添加数据或者购物车表修改数据并且修改的是购买数量字段而且更改后比更改前高才是加购
 *
 */
public class DwdTradeCartAdd extends BaseSQLApp {

    public static void main(String[] args) {
        new DwdTradeCartAdd().start(10013,4,"wuyong");
    }

    @Override
    public void handle(StreamTableEnvironment streamTable) {
//        TODO 1.读取kafka中topic_db的数据
        readOdsTopicdb(streamTable);

//        TODO 2.过滤出加购的数据-------------购物车表添加数据或者购物车表修改数据并且修改的是购买数量字段而且更改后比更改前高才是加购
        Table table = streamTable.sqlQuery("select  `data`['id'] id,\n" +
                "`data`['user_id'] user_id,\n" +
                "`data`['sku_id'] sku_id,\n" +
                "if(type='insert',cast(`data`['sku_num'] as string),cast((cast(data['sku_num'] as int ) - cast(`old`['sku_num'] as int )) as string)) sku_num  ,\n" +
                "ts\n" +
                "from KafkaTable\n" +
                "where `table`= 'cart_info'\n" +
                "and type='insert' or (type='update' and `old`['sku_num'] is not null and (cast(`old`['sku_num'] as int ) < cast(data['sku_num'] as int )))");

//        table.execute().print();

//        TODO 3.往kafka中写入数据
         streamTable.executeSql("CREATE TABLE "+Constant.TOPIC_DWD_TRADE_CART_ADD+" (\n" +
                "  id STRING,\n" +
                "  user_id STRING,\n" +
                "  sku_id STRING,\n" +
                "  sku_num STRING,\n" +
                "  ts bigint,\n" +
                "  PRIMARY KEY (id) NOT ENFORCED\n" +
                ") " + SQLUtil.getUpdateKafkaDDL(Constant.TOPIC_DWD_TRADE_CART_ADD));

        table.executeInsert(Constant.TOPIC_DWD_TRADE_CART_ADD);

    }
}
